package com.run.fjy.mr;

import java.io.IOException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.run.bcpimp.builder.IntegerValueBuilder;
import com.run.bcpimp.builder.LongValueBuilder;
import com.run.bcpimp.builder.StringValueBuilder;
import com.run.bcpimp.conf.BcpExtractorParser;
import com.run.bcpimp.conf.DbFieldConfig;
import com.run.bcpimp.conf.DbTableConfig;
import com.run.bcpimp.exception.BcpParseException;
import com.run.bcpimp.mr.io.ExtractTableValueWritable;
import com.run.bcpimp.processor.ResourceFactory;
import com.run.bcpimp.util.BcpUtils;
import com.run.fjy.io.TerminalWritable;

/**
 * 把这次的数据去重后组织成 {@link TerminalWritable}
 * 输出到相应的HDFS目录下。
 * @author chenxu
 *
 */
public class NewAddLocalReducer extends Reducer<Text,ExtractTableValueWritable,Text,TerminalWritable>
{

	private static Logger logger = LoggerFactory.getLogger(NewAddLocalReducer.class);
	private List<DbFieldConfig> dbFieldsList;
	private Map<Text, Set<Text>> tempMap = new HashMap<Text, Set<Text>>();
	private Context context;


	protected void cleanup(Context context)
		throws IOException, InterruptedException
	{
	}

	protected void setup(Context context)
		throws IOException, InterruptedException
	{
		this.context = context;
		Configuration configuration = context.getConfiguration();
		ResourceFactory rf = new ResourceFactory();
		rf.setup(configuration);
		logger.info("setup resource-factory");
		BcpExtractorParser parser = new BcpExtractorParser(configuration.get("cfgfile.bcpextract"));
		logger.info("init bcp-extrqact-parser");
		Map<String, DbTableConfig> dbTableConfigMap = parser.getDbTableConfigMap();
		DbTableConfig dbTableConfig = (DbTableConfig)dbTableConfigMap.get("newadd");
		dbFieldsList = dbTableConfig.getDbFieldsList();
		logger.info("init dbField meta begin");
		
		for(DbFieldConfig dbFieldConfig : dbTableConfig.getDbFieldsList()) {
			logger.info((new StringBuilder()).append("name[").append(dbFieldConfig.name).append("], type[").append(dbFieldConfig.type).append("], index[").append(dbFieldConfig.colIndex).append("]").toString());
			if ("string".equals(dbFieldConfig.type) || "array".equals(dbFieldConfig.type))
				dbFieldConfig.valueBuilder = new StringValueBuilder();
			else
			if ("int".equals(dbFieldConfig.type))
				if (dbFieldConfig.len < 10)
					dbFieldConfig.valueBuilder = new IntegerValueBuilder();
				else
					dbFieldConfig.valueBuilder = new LongValueBuilder();
		}
		
		logger.info("init dbField meta end");
	}

	/**
	 * 将当前 key 的所有值收集去重后输出到HDFS
	 */
	protected void reduce(Text key, Iterable<ExtractTableValueWritable> value, Context context)
		throws IOException, InterruptedException
	{
		tempMap.clear();
		TerminalWritable outValue = new TerminalWritable();
		outValue.setLocal(true);
		
		for(ExtractTableValueWritable oneValue : value) {
			collectValue(oneValue, outValue);
		}
		context.write(key, outValue);
		
	}

	/**
	 * 处理当前key的一个value
	 * @param oneValue
	 * @param outValue
	 * @throws BcpParseException
	 */
	private void collectValue(ExtractTableValueWritable oneValue, TerminalWritable outValue)
		throws BcpParseException
	{
		if (oneValue.getLastTime() < outValue.getFirstTime())
			outValue.setFirstTime(oneValue.getLastTime());
		Text colValues = oneValue.getColValues();
		byte bytes[] = colValues.getBytes();
		if (bytes.length != colValues.getLength())
			context.getCounter(BcpUtils.MR_COUNTER_GROUP, "error: text.getLength != bytes.length").increment(1L);
		int length = colValues.getLength();
		int colIndex = 0;
		int pos = 0;
		int start = 0;
		int clen = 0;
		DbFieldConfig dbFieldConfig;
		for (; pos < length; pos++)
		{
			if (bytes[pos] != BcpUtils.EXTRACT_COLVALUES_SPLIT)
				continue;
			clen = pos - start;
			dbFieldConfig = (DbFieldConfig)dbFieldsList.get(colIndex);
			if (clen > dbFieldConfig.len)
				context.getCounter(BcpUtils.MR_COUNTER_GROUP, (new StringBuilder()).append("error: value len exceed --").append(dbFieldConfig.name).toString()).increment(1L);
			else
			if (clen > 0)
			{
				String valueOf = String.valueOf(dbFieldConfig.valueBuilder.build(bytes, start, clen));
				dealField(dbFieldConfig, valueOf, outValue);
			} else
			{
				context.getCounter(BcpUtils.MR_COUNTER_GROUP, (new StringBuilder()).append("error: value len ").append(clen).append(" --").append(dbFieldConfig.name).toString()).increment(1L);
			}
			start = pos + 1;
			colIndex++;
		}

		clen = length - start;
		dbFieldConfig = (DbFieldConfig)dbFieldsList.get(colIndex);
		if (clen > 0)
		{
			String valueOf = String.valueOf(dbFieldConfig.valueBuilder.build(bytes, start, clen));
			dealField(dbFieldConfig, valueOf, outValue);
		} else
		{
			return;
		}
	}

	/**
	 * array类型的字段取为一值，其他字段直接添加
	 * @param dbFieldConfig
	 * @param value
	 * @param outValue
	 */
	private void dealField(DbFieldConfig dbFieldConfig, String value, TerminalWritable outValue)
	{
		Text key = new Text(dbFieldConfig.name);
		Text valueText = new Text(value);
		if (dbFieldConfig.type.equals("array"))
		{
			Set<Text> set = (Set<Text>)tempMap.get(key);
			if (null == set)
			{
				set = new HashSet<Text>();
				tempMap.put(key, set);
			}
			if (set.add(valueText))
				outValue.addLocal(key, valueText);
		} else
		{
			outValue.getOtherFields().put(key, valueText);
		}
	}


}